package com.yanggu.flink.cep_api

import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.cep.functions.{PatternProcessFunction, TimedOutPartialMatchHandler}
import org.apache.flink.cep.scala.CEP
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

import java.util

/**
 * CEP中指定时间内, 没有按照数据结尾
 */
object CepDemo02 {

  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)

    val value = environment
      .fromElements(("zhangsan", "浏览商品", 100L))
      .assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps()
      .withTimestampAssigner(new SerializableTimestampAssigner[(String, String, Long)] {
        override def extractTimestamp(element: (String, String, Long), recordTimestamp: Long): Long = element._3
      }))
      .keyBy(_._1)

    val value1 = Pattern
      .begin[(String, String, Long)]("begin").where(tuple => tuple._2.equals("浏览商品"))
      .notFollowedBy("end").where(tuple => tuple._2.equals("下单"))
      .within(Time.seconds(10L))

    val value2 = CEP
      .pattern(value, value1)
      .process(new PatternProcessFunction[(String, String, Long), String] with TimedOutPartialMatchHandler[(String, String, Long)] {
        override def processMatch(tempMatch: util.Map[String, util.List[(String, String, Long)]], ctx: PatternProcessFunction.Context, out: Collector[String]) = {
          val tuple = tempMatch.get("begin").get(0)
          out.collect(s"${tuple._1}在10s内未支付")
        }

        override def processTimedOutMatch(tempMatch: util.Map[String, util.List[(String, String, Long)]], ctx: PatternProcessFunction.Context) = {
          val tuple = tempMatch.get("begin").get(0)
          ctx.output(OutputTag[String]("test"), (s"${tuple._1}在10s内未支付"))
        }
      })
    value2.print("test1>>>>>")
    value2.getSideOutput(OutputTag[String]("test")).print("test2>>>>>>")

    environment.execute()

  }

}
